package com.eshore.mq.Mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * @author ZhouXiong
 * @Description
 * @apiDefine Created by ZhouX on 2021/9/6
 */
@Service
@Slf4j
public class MqttProducer {

    @Autowired
    MqttClient mqttClient;

    /**
     * 向某个主题发布消息 默认qos：1
     *
     * @param topic:发布的主题
     * @param msg：发布的消息
     */
    public void pub(String topic, String msg) throws MqttException {
        log.info("----------开始推送消息，目标主题：{}，消息：{}", topic, msg);
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setPayload(msg.getBytes());
        MqttTopic mqttTopic = mqttClient.getTopic(topic);
        MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
        token.waitForCompletion();
        log.info("----------消息推送成功，目标主题：{}，消息：{}", topic, msg);
    }

    /**
     * 向某个主题发布消息
     *
     * @param topic: 发布的主题
     * @param msg:   发布的消息
     * @param qos:   消息质量    Qos：0、1、2
     */
    public void pub(String topic, String msg, int qos) throws MqttException {
        log.info("----------开始推送消息，目标主题：{}，消息：{}", topic, msg);
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(qos);
        mqttMessage.setPayload(msg.getBytes());
        MqttTopic mqttTopic = mqttClient.getTopic(topic);
        MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
        token.waitForCompletion();
        log.info("----------消息推送成功，目标主题：{}，消息：{}", topic, msg);
    }

    /**
     * 订阅某一个主题 ，此方法默认的的Qos等级为：1
     */
    public void sub(String topic) throws MqttException {
        log.info("----------开始订阅主题----------");
        IMqttToken iMqttToken = mqttClient.subscribeWithResponse(topic);
        byte[] payload = iMqttToken.getResponse().getPayload();
    }

}
